package com.book.flink

import com.alibaba.fastjson.JSON
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.http.client.methods.HttpPost
import org.apache.http.entity.StringEntity
import org.apache.http.impl.client.HttpClients
import org.apache.http.util.EntityUtils

class HttpSinkFunction extends RichSinkFunction[String]{

//  val postUrl = "http://web.books.com/books/api/web/index.php/streaming/kafka"
  val postUrl = "https://lq.yifengshuyuan.com/books/api/web/index.php/streaming/kafka"

  def postResponse(url: String, params: String = null, header: String = null) : String = {
    val httpClient = HttpClients.createDefault()
    val post = new HttpPost(url)

    if (null != header) {
      val json = JSON.parseObject(header)
      json.keySet().toArray().map(_.toString).foreach(key => post.setHeader(key, json.getString(key)))
    }

    if (null != params) {
      post.setEntity(new StringEntity(params, "UTF-8"))
    }
    val response = httpClient.execute(post)
    var data = EntityUtils.toString(response.getEntity, "UTF-8")
    httpClient.close()
    data
  }

  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {
    var result = postResponse(postUrl, value,"""{"Content-Type": "application/json"}""")

    println(result)
  }

  override def close(): Unit = super.close()

  override def open(parameters: Configuration): Unit = super.open(parameters)
}
